#include "config.h"

#include <sys/types.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <dirent.h>
#include <unistd.h>

#define DBG_SUBSYS S_LIBSTORAGE

#include "limits.h"
#include "adt.h"
#include "sysy_lib.h"
#include "bmap.h"
#include "net_table.h"
#include "configure.h"
#include "table_proto.h"
#include "volume_proto.h"
#include "metadata.h"
#include "md_proto.h"
#include "md_map.h"
#include "stor_ctl.h"
#include "rmsnap_bh.h"
#include "core.h"
#include "volume_bh.h"
#include "volume_ctl.h"
#include "volume.h"
#include "rmvol_bh.h"
#include "../storage/stor_rpc.h"
#include "../storage/md_parent.h"
#include "lich_md.h"
#include "../replica/replica.h"
#include "net_global.h"
#include "job_dock.h"
#include "ylog.h"
#include "dbg.h"

#include "lsv_conf.h"
#include "lsv_bitmap.h"
#include "lsv_bitmap_snap.h"
#include "lsv_wbuffer.h"
#include "lsv_rcache.h"
#include "pool_list.h"

static int __volume_proto_snapshot_listopen(volume_proto_t *volume_proto, const char *uuid)
{
        int ret, fd;
        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;

        ret = pool_list_newfd(&fd);
        if (unlikely(ret))
                UNIMPLEMENTED(__DUMP__);

        lsv_bitmap_list_snap(lsv_info, fd);

        ret = pool_list_addfd(uuid, fd);
        if (unlikely(ret))
                UNIMPLEMENTED(__DUMP__);

        return 0;
}

static int __volume_proto_snapshot_list(volume_proto_t *volume_proto, const char *uuid, uint64_t offset, void *de, int *delen)
{
        (void) volume_proto;
        return pool_list_lspool(uuid, offset, de, delen);
}

static int __volume_proto_snapshot_listclose(volume_proto_t *volume_proto, const char *uuid)
{
        (void) volume_proto;
        return pool_list_closefd(uuid);
}
static int __volume_proto_snapshot_create(volume_proto_t *volume_proto,
                                          const char *name, int p, const char *_site, int force)
{
        int ret;
        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;
        table1_t *table1 = &volume_proto->table1;
        volume_format_t format = lich_volume_format(&table1->fileinfo);

        (void)p;//???????????????????????????????;//???????????????????????????????;//???????????????????????????????;//???????????????????????????????
        (void) _site;
        (void) force;
        
        DINFO("__volume_proto_snapshot_create begin\r\n");

        // @malloc ERR1
        if(format == VOLUME_FORMAT_LSV)
        {
                lsv_wrlock(&lsv_info->io_lock);

                lsv_wbuffer_flush_unlock(lsv_info);
        }
        else if(format == VOLUME_FORMAT_ROW2 || format == VOLUME_FORMAT_ROW3)
        {
                lsv_wrlock(&lsv_info->io_lock);
        }

        DINFO("__volume_proto_snapshot_create->lsv_wbuffer_flush end\r\n");

        ret = lsv_bitmap_create_snapshot(lsv_info, NULL, name, NULL);
        if (unlikely(ret)) {
                GOTO(ERR1, ret);
        }

        if(format == VOLUME_FORMAT_LSV)
        {
                lsv_unlock(&lsv_info->io_lock);
        }
        else if(format == VOLUME_FORMAT_ROW2 || format == VOLUME_FORMAT_ROW3)
        {
                lsv_unlock(&lsv_info->io_lock);
        }

        DINFO("__volume_proto_snapshot_create->__volume_proto_snapshot_create end\r\n");
        DINFO("__volume_proto_snapshot_create end\r\n");
        return 0;
ERR1:

        if(format == VOLUME_FORMAT_LSV)
        {
                lsv_unlock(&lsv_info->io_lock);
        }
        else if(format == VOLUME_FORMAT_ROW2 || format == VOLUME_FORMAT_ROW3)
        {
                lsv_unlock(&lsv_info->io_lock);
        }

        return ret;
}

static int __volume_proto_snapshot_remove(volume_proto_t *volume_proto,
                                          const char *name, int force)
{
        int ret;
        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;
        table1_t *table1 = &volume_proto->table1;
        volume_format_t format = lich_volume_format(&table1->fileinfo);

        (void) force;
        DINFO("__volume_proto_snapshot_remove begin\r\n");

        if(format == VOLUME_FORMAT_LSV)
        {
                // @malloc ERR1
                lsv_wrlock(&lsv_info->io_lock);

                lsv_wbuffer_flush_unlock(lsv_info);
                lsv_wbuffer_clean_unlock(lsv_info);

                DINFO("__volume_proto_snapshot_remove->lsv_wbuffer_clean end\r\n");
                //clean up rcache.
                ret = lsv_rcache_clean(lsv_info, LSV_CHUNK_NULL);
                if(unlikely(ret)){
                        GOTO(err_ret, ret);
                }
        }

        DINFO("__volume_proto_snapshot_remove->lsv_rcache_clean end\r\n");
        ret = lsv_bitmap_delete_snapshot(lsv_info, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if(format == VOLUME_FORMAT_LSV)
        {
                lsv_unlock(&lsv_info->io_lock);
        }

        DINFO("__volume_proto_snapshot_remove->lsv_bitmap_delete_snapshot end\r\n");

        DINFO("__volume_proto_snapshot_remove end\r\n");
        return 0;
err_ret:
        if(format == VOLUME_FORMAT_LSV)
                lsv_unlock(&lsv_info->io_lock);

        return ret;
}

static int __volume_proto_snapshot_rollback(volume_proto_t *volume_proto, const char *name)
{
 	int ret;
        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;
        table1_t *table1 = &volume_proto->table1;
        volume_format_t format = lich_volume_format(&table1->fileinfo);

        DINFO("__volume_proto_snapshot_rollback begin\r\n");

        if(format == VOLUME_FORMAT_LSV)
        {
                // @malloc ERR1
                lsv_wrlock(&lsv_info->io_lock);

                lsv_wbuffer_flush_unlock(lsv_info);
                lsv_wbuffer_clean_unlock(lsv_info);

                DINFO("__volume_proto_snapshot_rollback->lsv_wbuffer_clean end\r\n");
                //clean up rcache.
                ret = lsv_rcache_clean(lsv_info, LSV_CHUNK_NULL);
                if(unlikely(ret)){
                        GOTO(err_ret, ret);
                }
        }
        DINFO("__volume_proto_snapshot_rollback->lsv_rcache_clean end\r\n");
        ret = lsv_bitmap_revert_snapshot(lsv_info, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if(format == VOLUME_FORMAT_LSV)
                lsv_unlock(&lsv_info->io_lock);

        DINFO("__volume_proto_snapshot_rollback->lsv_bitmap_revert_snapshot end\r\n");

        DINFO("__volume_proto_snapshot_rollback end\r\n");
        return 0;
err_ret:
        if(format == VOLUME_FORMAT_LSV)
                lsv_unlock(&lsv_info->io_lock);
        return ret;
}

static int __volume_proto_snapshot_lookup(volume_proto_t *volume_proto, const char *name, chkinfo_t *chkinfo)
{
        (void) volume_proto;
        (void) name;
        (void) chkinfo;
        return 0;
}

static int __volume_proto_snapshot_check(volume_proto_t *volume_proto, const char *name)
{
        (void) volume_proto;
        (void) name;
        return 0;
}

static int __volume_proto_snapshot_protect(volume_proto_t *volume_proto, const snap_protect_param_t on)
{
        int ret;
        DINFO("__volume_proto_snapshot_protect %d\r\n", on.on);

        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;

        //todo.
        ret = lsv_bitmap_protect_snapshot(lsv_info, on.snap_name, on.on);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __volume_proto_snapshot_last(volume_proto_t *volume_proto,
                nid_t *nid, fileid_t *fileid, char *name, uint64_t *snap_version)
{
        // 若返回0，影响cleanup过程
        (void) volume_proto;
        (void) nid;
        (void) fileid;
        (void) name;
        (void) snap_version;
        return ENOENT;
}

#if 0
static int __volume_proto_snapshot_cleanup(volume_proto_t *volume_proto)
{
        (void) volume_proto;
        return 0;
}

static int __volume_proto_snapshot_newchunk(const nid_t *snapnid,
                                            const fileid_t *snapid, int idx, const buffer_t *buf)
{
        (void) snapnid;
        (void) snapid;
        (void) idx;
        (void) buf;
        return 0;
}

static int __volume_proto_snapshot_cow__(volume_proto_t *volume_proto,
                                       const chkid_t *chkid, const nid_t *snapnid,
                                       const fileid_t *snapid)
{
        (void) volume_proto;
        (void) chkid;
        (void) snapnid;
        (void) snapid;
        return 0;
}

static void __volume_proto_snapshot_read_remote__(void *_core_ctx)
{
        (void) _core_ctx;
}

static int __volume_proto_snapshot_rollback__(volume_proto_t *volume_proto,
                                              const nid_t *snapnid, const fileid_t *snapid,
                                              const chkid_t *chkid)
{
        (void) volume_proto;
        (void) snapnid;
        (void) snapid;
        (void) chkid;
        return 0;
}

static int __volume_proto_snapshot_cow(volume_proto_t *volume_proto,
                                     const chkid_t *chkid)
{
        (void) volume_proto;
        (void) chkid;
        return 0;
}

static int __volume_proto_snapshot_chunk_rollback(volume_proto_t *volume_proto, const chkid_t *chkid)
{
        (void) volume_proto;
        (void) chkid;
        return 0;
}
#endif

static int __volume_proto_snapshot_rollback_bh1(volume_proto_t *volume_proto, chkid_t *chkid)
{
        (void) volume_proto;
        (void) chkid;
        return 0;
}

#if 0
static int __volume_proto_snapshot_rollback_bh2(volume_proto_t *volume_proto, fileid_t *snapid)
{
        (void) volume_proto;
        (void) snapid;
        return 0;
}
#endif

static int __volume_proto_snapshot_rollback_bh(volume_proto_t *volume_proto, buffer_t *buf)
{
        (void) volume_proto;
        (void) buf;
        return 0;
}

#if 0
static int __volume_proto_snapshot_cleanup_chunk1(const nid_t *srcnid,
                                                  const fileid_t *srcid, const nid_t *distnid,
                                                  const fileid_t *distid, int idx, uint64_t max)
{
        (void) srcnid;
        (void) srcid;
        (void) distnid;
        (void) distid;
        (void) idx;
        (void) max;
        return 0;
}

static int __volume_proto_snapshot_chunk_exist(const nid_t *snapnid,
                                            const fileid_t *snapid, int idx, int *exist)
{
        (void) snapnid;
        (void) snapid;
        (void) idx;
        (void) exist;
        return 0;
}
#endif

static int __volume_proto_snapshot_cleanup_chunk(const nid_t *srcnid,
                                                 const fileid_t *srcid, const nid_t *distnid,
                                                 const fileid_t *distid, int idx, uint64_t max,
                                                 const ec_t *ec)
{
        (void) srcnid;
        (void) srcid;
        (void) distnid;
        (void) distid;
        (void) idx;
        (void) max;
        (void) ec;
        return 0;
}

#if 0
static int __volume_proto_snapshot_cleanup_bh2__(const chkinfo_t*chkinfo)
{
        (void) chkinfo;
        return 0;
}
#endif

static int __volume_proto_snapshot_cleanup_bh2(volume_proto_t *volume_proto, const char *name)
{
        (void) volume_proto;
        (void) name;
        return 0;
}

#if 0
static void __volume_proto_snapshot_chunk_read_remote__(void *_core_ctx)
{
        (void) _core_ctx;
}

static int __volume_proto_snapshot_chunk_read_remote(const fileid_t *parent, const fileid_t *fileid,
                buffer_t *buf, size_t size, off_t offset)
{
        (void) parent;
        (void) fileid;
        (void) buf;
        (void) size;
        (void) offset;
        return 0;
}
#endif

static int __volume_proto_snapshot_read(volume_proto_t *volume_proto, const io_t *io, buffer_t *buf)
{
        lsv_io_t *lsv_io = (lsv_io_t *)io;

        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;

        DINFO("__volume_proto_snapshot_read\r\n");

        raw_io_t rio;
        raw_io_init(&rio, io->offset, io->size);

        int rc = lsv_rcache_lookup(lsv_info, lsv_io->chunk_id, lsv_io->chunk_off / LSV_PAGE_SIZE * LSV_PAGE_SIZE,
                        lsv_io->chunk_off, io->size, &rio, buf);
        if (rc < 0) {
                DERROR("rcache_lookup err.errno:%d\n", rc);
                return rc;
        }

        if (rc < lsv_io->size) {
                mbuffer_appendzero(buf, lsv_io->size - rc);
                rc = lsv_io->size;

                DINFO("__volume_proto_snapshot_read end %d\r\n", rc);

                return 0;
        }

        DINFO("__volume_proto_snapshot_read end %d\r\n", 0);

        return 0;
}


static int __volume_proto_snapshot_read_meta(volume_proto_t *volume_proto, const io_t *io, const char *snap, buffer_t *buf)
{
        DINFO("__volume_proto_snapshot_read_meta\r\n");

        lsv_io_t *lsv_io = (lsv_io_t *)io;
        void *_buf = mem_cache_calloc(MEM_CACHE_8K, lsv_io->size);
        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;

        int rc = lsv_bitmap_read_snapshot_meta(lsv_info, snap, io->offset, io->size, _buf);

        if(!rc)
                mbuffer_appendmem(buf, _buf, io->size);

        mem_cache_free(MEM_CACHE_8K, _buf);

        DINFO("__volume_proto_snapshot_read_meta %d\r\n", rc);

        return rc;
}


static int __volume_proto_lower_read(volume_proto_t *volume_proto, const io_t *io, buffer_t *buf)
{
        lsv_io_t *lsv_io = (lsv_io_t *)io;

        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;

        DINFO("__volume_proto_lower_read %llu, %u, %u,\r\n", (LLU)lsv_io->chunk_id, lsv_io->chunk_off,
              lsv_io->size);

        void *_buf = mem_cache_calloc(MEM_CACHE_8K, lsv_io->size);

        lsv_s32_t ret = lsv_volume_chunk_read_data(lsv_info, 0, lsv_io->chunk_id, lsv_io->chunk_off,
                                             lsv_io->size, _buf);

        if(!ret)
                mbuffer_appendmem(buf, _buf, lsv_io->size);

        mem_cache_free(MEM_CACHE_8K, _buf);

        DINFO("__volume_proto_lower_read end %d\r\n", 0);

        return ret;
}

int row2_volume_proto_snapshot_flat(volume_proto_t *volume_proto, int idx)
{
        lsv_volume_proto_t *lsv_info = &volume_proto->table1.lsv_info;

        (void) idx;
        DINFO("row2_volume_proto_snapshot_flat begin %d\r\n", 0);

        int ret = lsv_bitmap_snap_full_clone(lsv_info);

        DINFO("row2_volume_proto_snapshot_flat end %d\r\n", 0);

        return ret;
}

int volume_proto_snapshot_lsv_init(volume_proto_t *volume_proto)
{
        volume_proto->snapshot_create = __volume_proto_snapshot_create;
        volume_proto->snapshot_listopen = __volume_proto_snapshot_listopen;
        volume_proto->snapshot_list = __volume_proto_snapshot_list;
        volume_proto->snapshot_listclose = __volume_proto_snapshot_listclose;
        volume_proto->snapshot_lookup = __volume_proto_snapshot_lookup;
        volume_proto->snapshot_check = __volume_proto_snapshot_check;
        volume_proto->snapshot_remove = __volume_proto_snapshot_remove;
        volume_proto->snapshot_protect = __volume_proto_snapshot_protect;
        volume_proto->snapshot_last = __volume_proto_snapshot_last;
        volume_proto->snapshot_rollback = __volume_proto_snapshot_rollback;
        volume_proto->snapshot_rollback_bh1 = __volume_proto_snapshot_rollback_bh1;
        // volume_proto->snapshot_rollback_bh2 = __volume_proto_snapshot_rollback_bh2;
        volume_proto->snapshot_rollback_bh = __volume_proto_snapshot_rollback_bh;
        //volume_proto->snapshot_cleanup = __volume_proto_snapshot_cleanup;
        volume_proto->snapshot_cleanup_chunk = __volume_proto_snapshot_cleanup_chunk;
        volume_proto->snapshot_cleanup_bh2 = __volume_proto_snapshot_cleanup_bh2;
        volume_proto->snapshot_read = __volume_proto_snapshot_read;
        volume_proto->snapshot_read_meta = __volume_proto_snapshot_read_meta;
        volume_proto->lower_read = __volume_proto_lower_read;
        volume_proto->snapshot_flat = row2_volume_proto_snapshot_flat;

        uuid_clear(volume_proto->snapshot_lock_uuid);

        return 0;
}
